//
// Created by lz on 2021/2/17.
//

#include "net/EventLoop.h"

#include "base/Logging.h"
#include "base/Mutex.h"
#include "net/Channel.h"
#include "net/SocketsOps.h"
#include "net/TimerQueue.h"
#include "net/Poller.h"

#include <algorithm>
#include <csignal>
#include <sys/eventfd.h>
#include <unistd.h>

using namespace lzweb;
using namespace lzweb::net;

namespace
{
	__thread EventLoop* t_loopInThisThread = 0;

	const int kPollTimeMs = 10000;

	int createEventfd()
	{
		int evtfd = ::eventfd(0, EFD_NONBLOCK | EFD_CLOEXEC);
		if (evtfd < 0)
		{
			LOG_SYSERR << "Failed in eventfd";
			abort();
		}

		return evtfd;
	}

#pragma GCC diagnostic ignored "-Wold-style-cast"

	class IgnoreSigPipe
	{
	 public:
		IgnoreSigPipe()
		{
			//一旦使用，相当向进程注册一个这样的信号及信号处理函数。（如无修改整个进程生命周期都有效）
			::signal(SIGPIPE, SIG_IGN);
		}
	};
#pragma GCC diagnostic error "-Wold-style-cast"

	//这个变量的声明及默认初始化会使得程序一开始，SIGPIPE就会按SIG_IGN处理？
	IgnoreSigPipe initObj;
} // namespace

EventLoop* EventLoop::getEventLoopOfCurrentThread()
{
	return t_loopInThisThread;
}

EventLoop::EventLoop()
	: looping_(false),
	  quit_(false),
	  eventHandling_(false),
	  callingPendingFunctors_(false),
	  iteration_(0),
	  threadId_(CurrentThread::tid()),
	  poller_(Poller::newDefaultPoller(this)),
	  timerQueue_(new TimerQueue(this)),
	  wakeupFd_(createEventfd()),
	  wakeupChannel_(new Channel(this, wakeupFd_)),
	  currentActiveChannel_(NULL)
{
	LOG_DEBUG << " EventLoop created " << this << " in thread " << threadId_;
	if(t_loopInThisThread)
	{
		LOG_FATAL << "Another EventLoop" << t_loopInThisThread
				<<" exists in this thread " << threadId_;
	}
	else
	{
		t_loopInThisThread = this;
	}

	wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead, this));
	// we are always reading the wakeupfd
	wakeupChannel_->enableReading();
}

EventLoop::~EventLoop()
{
	LOG_DEBUG << " EventLoop " << this << " of thread " << threadId_
				<< " destructs in thread " << CurrentThread::tid();

	wakeupChannel_->disableAll();
	wakeupChannel_->remove();
	::close(wakeupFd_);
	t_loopInThisThread = NULL;
}

void EventLoop::loop()
{
	assert(!looping_);
	assertInLoopThread();
	// 开始 loop
	looping_ = true;
	quit_ = false; // FIXME: what if someone calls quit() before loop() ? (所以就加了这一句？)

	LOG_TRACE << "EventLoop " << this <<" start looping";

	while(!quit_)
	{
		activeChannels_.clear();
		pollReturnTime_ = poller_->poll(kPollTimeMs, &activeChannels_);
		++iteration_;
		if(Logger::logLevel() <= Logger::TRACE)
		{
			printActiveChannels();
		}
		// TODO sort channel by priority
		eventHandling_ = true;

		//socket fd的触发事件及其处理函数都在该线程处理？
		for(Channel* channel : activeChannels_)
		{
			currentActiveChannel_ = channel;
			currentActiveChannel_->handleEvent(pollReturnTime_);
		}

		currentActiveChannel_ = NULL;
		eventHandling_ = false;
		doPendingFunctors();
	}

	LOG_TRACE << "EventLoop " << this << " stop looping";
	looping_ = false;
}

void EventLoop::quit()
{
	quit_ = true;
	// There is a chance that loop() just executes while(!quit_) and exits,
	// then EventLoop destructs, then we are accessing an invalid object.
	// Can be fixed using mutex_ in both places.
	if(! isInLoopThread())
	{
		wakeup();
	}
}

//如果在所属线程，则直接运行，否则添加到队列中
void EventLoop::runInLoop(EventLoop::Functor cb)
{
	if(isInLoopThread())
	{
		cb();
	}
	else
	{
		queueInLoop(std::move(cb));
	}
}
//把cb 函数扔进EventLoop的队列中
void EventLoop::queueInLoop(EventLoop::Functor cb)
{
	{
		MutexLockGuard lock(mutex_);
		pendingFunctors_.push_back(std::move(cb));
	}
	if (!isInLoopThread() || callingPendingFunctors_)
	{
		wakeup();
	}
}

size_t EventLoop::queueSize() const
{
	MutexLockGuard lock(mutex_);
	return pendingFunctors_.size();
}


// 根据时间点 运行
TimerId EventLoop::runAt(TimeStamp time, TimerCallback cb)
{
	return timerQueue_->addTimer(std::move(cb), time, 0.0);
}

// 根据延长时间段运行
TimerId EventLoop::runAfter(double delay, TimerCallback cb)
{
	TimeStamp time(addTime(TimeStamp::now(), delay));
	return runAt(time, cb);
}

TimerId EventLoop::runEvery(double interval, TimerCallback cb)
{
	TimeStamp time(addTime(TimeStamp::now(), interval));
	return timerQueue_->addTimer(cb, time, interval);
}

void EventLoop::cancel(TimerId timerId)
{
	return timerQueue_->cancel(timerId);

}

void EventLoop::updateChannel(Channel* channel)
{
	assert(channel->ownerLoop() == this);
	assertInLoopThread();
	poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
	assert(channel->ownerLoop() == this);
	assertInLoopThread();
	if(eventHandling_)
	{
		//question: 下面的第二个条件是不是填错了
		assert(currentActiveChannel_ == channel ||
			   std::find(activeChannels_.begin(), activeChannels_.end(), channel) == activeChannels_.end());
	}
	poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel)
{
	assert(channel->ownerLoop() == this);
	assertInLoopThread();
	return poller_->hasChannel(channel);
}

void EventLoop::abortNotInLoopThread()
{
	LOG_FATAL << "EventLoop::abortNotInLoopThread - EventLoop " << this
				<< " was created in threadId_ = " << threadId_
				<<" , current thread id = " << CurrentThread::tid();
}

void EventLoop::wakeup()
{
	uint64_t  one = 1;
	ssize_t n = sockets::write(wakeupFd_, &one, sizeof one);
	if(n != sizeof one)
	{
		LOG_ERROR << "EventLoop::handleRead() writes " << n << " bytes instead of 8";
	}

}

void EventLoop::handleRead()
{
	uint64_t  one = 1;
	ssize_t n = sockets::read(wakeupFd_, &one, sizeof one);
	if (n != sizeof one)
	{
		LOG_ERROR << "EventLoop::handleRead() reads " << n << " bytes instead of 8";
	}
}

void EventLoop::doPendingFunctors()
{
	//
	std::vector<Functor> functors;
	callingPendingFunctors_ = true;

	{
		MutexLockGuard lock(mutex_);
		functors.swap(pendingFunctors_);
	}

	//PendingFunctors中的处理函数也会在EventLoop所在的线程
	for(const Functor& functor : functors)
	{
		functor();
	}

	callingPendingFunctors_ = false;
}

void EventLoop::printActiveChannels() const
{
	for (const Channel* channel : activeChannels_)
	{
		LOG_TRACE << "{" << channel->reventsToString() << "}";
	}

}




